package com.dt.platform.ops.service.impl;

import com.alibaba.csp.sentinel.util.StringUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dt.platform.constants.enums.ops.MonitorMethodEnum;
import com.dt.platform.domain.ops.MonitorNode;
import com.dt.platform.domain.ops.MonitorTplIndicator;
import com.dt.platform.ops.service.IMonitorDataProcessJdbcTplService;
import com.github.foxnic.api.error.ErrorDesc;
import com.github.foxnic.api.transter.Result;
import com.github.foxnic.commons.busi.id.IDGenerator;
import com.github.foxnic.commons.concurrent.SimpleJoinForkTask;
import com.github.foxnic.commons.encrypt.AESUtil;
import com.github.foxnic.commons.log.Logger;
import com.github.foxnic.dao.data.RcdSet;
import com.github.foxnic.dao.spec.DAO;
import com.github.foxnic.sql.expr.Insert;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.github.foxnic.web.framework.dao.DBConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Service("MonitorDataProcessJdbcTplServiceImpl")
public class MonitorDataProcessJdbcTplServiceImpl implements IMonitorDataProcessJdbcTplService {
    /**
     * 注入DAO对象
     * */
    @Resource(name= DBConfigs.PRIMARY_DAO)
    private DAO dao=null;


    /**
     * 获得 DAO 对象
     * */
    public DAO dao() { return dao; }


    @Autowired
    private MonitorDataProcessBaseServiceImpl monitorDataProcessBaseService;

    @Override
    public Result collectData() {
        List<MonitorNode> nodeList=new ArrayList<>();
        List<MonitorNode> nodeListMySQL5=monitorDataProcessBaseService.queryExecuteNodeList( MonitorMethodEnum.MYSQL_5_JDBC_TPL.code());
        List<MonitorNode> nodeListMySQL8=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.MYSQL_8_JDBC_TPL.code());
        List<MonitorNode> nodeListObOracle=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.OB_ORACLE_JDBC_TPL.code());
        List<MonitorNode> nodeListObMySQL=monitorDataProcessBaseService.queryExecuteNodeList( MonitorMethodEnum.OB_MYSQL_JDBC_TPL.code());
        List<MonitorNode> nodeListGaussDB=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.GAUSSDB_JDBC_TPL.code());
        List<MonitorNode> nodeListOracleDB=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.ORALCE_JDBC_TPL.code());


        if(nodeListMySQL5!=null&&nodeListMySQL5.size()>0){
            nodeList.addAll(nodeListMySQL5);
        }
        if(nodeListMySQL8!=null&&nodeListMySQL8.size()>0){
            nodeList.addAll(nodeListMySQL8);
        }
        if(nodeListObOracle!=null&&nodeListObOracle.size()>0){
            nodeList.addAll(nodeListObOracle);
        }
        if(nodeListObMySQL!=null&&nodeListObMySQL.size()>0){
            nodeList.addAll(nodeListObMySQL);
        }
        if(nodeListGaussDB!=null&&nodeListGaussDB.size()>0){
            nodeList.addAll(nodeListGaussDB);
        }
        if(nodeListOracleDB!=null&&nodeListOracleDB.size()>0){
            nodeList.addAll(nodeListOracleDB);
        }


        Logger.info("collectData,find node number:"+nodeList.size());
        // 创建 ForkJoin 工具，其中 输入一个 Integer 元素的 List ，输出 Long 元素的 List ，每个线程处理 若干元素 ，此处为 5 个
        if(nodeList.size()>0) {
            int batchSize=monitorDataProcessBaseService.calBathSizeValue(nodeList.size());
            Logger.info("current batch size:"+batchSize);
            SimpleJoinForkTask<MonitorNode, Result> task = new SimpleJoinForkTask<>(nodeList, batchSize);
            // 并行执行
            List<Result> rvs = task.execute(els -> {
                // 打印当前线程信息
                Logger.info(Thread.currentThread().getName());
                // 处理当前分到的 若干元素，此处为 5 个
                List<Result> rs = new ArrayList<>();
                for (MonitorNode node : els) {
                    Result result = collectNodeData(node, null, false);
                    if(result.isSuccess()){
                        Logger.debug("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }else{
                        Logger.error("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }
                    rs.add(result);
                }
                // 处理完毕，返回本批次的处理结果，由 SimpleJoinForkTask 负责合并全部结果，并由 rvs 变量接收
                return rs;
            });
        }
        return ErrorDesc.success();
    }
    //同一个节点指标必须一样
    @Override
    public Result collectNodeData(MonitorNode node,String indicatorId,boolean isForce) {
        String sql="select distinct c.monitor_method from ops_monitor_tpl a,ops_monitor_node_tpl_item b,ops_monitor_tpl_indicator c\n" +
                "where a.code=b.tpl_code \n" +
                "and a.code=c.monitor_tpl_code\n" +
                "and a.deleted=0\n" +
                "and c.deleted=0\n" +
                "and a.status='enable'\n" +
                "and c.status='enable'\n" +
                "and c.monitor_method in ('oracle_jdbc_tpl','ob_mysql_jdbc_tpl','ob_oracle_jdbc_tpl','mysql_5_jdbc_tpl','mysql_8_jdbc_tpl','gaussdb_jdbc_tpl')\n" +
                "and b.node_id=?";
        dao.pausePrintThreadSQL();
        RcdSet rs=dao.query(sql,node.getId());
        dao.resumePrintThreadSQL();
        if(rs.size()==0){
            return ErrorDesc.success();
        }
        if(rs.size()!=1){
            return ErrorDesc.failureMessage("当前节点,存在超过2个以上jdbc类型,只支持一个");
        }
        String var=node.getVar();
        JSONObject varObj=JSONObject.parseObject(var);
        if(var==null){
            return ErrorDesc.failureMessage("当前节点,所需的变量不存在");
        }
        String user=varObj.getString("user");
        String pwd=varObj.getString("pwd");
        if(varObj.containsKey("user")&&varObj.containsKey("pwd")){
            if(varObj.containsKey("encrypt")&& "true".equals(varObj.getString("encrypt"))){
                AESUtil aes=new AESUtil("collect");
                pwd=aes.decryptData(varObj.getString("pwd"));
            }
        }else{
            Logger.info("请输入连接的账户、密码信息");
            return ErrorDesc.failureMessage("请输入连接的账户、密码信息");
        }
        String driver="";
        String me=rs.getRcd(0).getString("monitor_method");
        List<MonitorTplIndicator> monitorTplIndicatorList=monitorDataProcessBaseService.queryExecuteIndicatorList(node.getId(),me,null,isForce);
        if(monitorTplIndicatorList==null||monitorTplIndicatorList.size()==0){
            Logger.info("collect method:jdbc_tpl,node:"+node.getNodeNameShow()+",当前没有指标触发采集");
            return ErrorDesc.success();
        }
        if(MonitorMethodEnum.OB_ORACLE_JDBC_TPL.code().equals(me)||MonitorMethodEnum.OB_ORACLE_JDBC_TPL.code().equals(me)){
            driver="com.oceanbase.jdbc.Driver";
        }else if(MonitorMethodEnum.GAUSSDB_JDBC_TPL.code().equals(me)){
            driver="com.huawei.gaussdb.jdbc.Driver";
        }else if(MonitorMethodEnum.MYSQL_5_JDBC_TPL.code().equals(me)){
            driver="com.mysql.jdbc.Driver";
        }else if(MonitorMethodEnum.MYSQL_8_JDBC_TPL.code().equals(me)){
            driver="com.mysql.cj.jdbc.Driver";
        }else if(MonitorMethodEnum.ORALCE_JDBC_TPL.code().equals(me)) {
            driver = "oracle.jdbc.driver.OracleDriver";
        }else{
            return ErrorDesc.failureMessage("当前节点所选JDBC不支持");
        }
        if(varObj.containsKey("driver")&& ( !StringUtil.isBlank(varObj.getString("driver")) )){
            driver=varObj.getString("driver");
        }
        Logger.debug("java.net.preferIPv4Stack: " + System.getProperty("java.net.preferIPv4Stack"));
        Logger.debug("java.net.preferIPv6Addresses: " + System.getProperty("java.net.preferIPv6Addresses"));
        Logger.debug("jdk.net.preferIPv4Stack: " + System.getProperty("jdk.net.preferIPv4Stack"));
        Logger.debug("jdk.net.preferIPv6Addresses: " + System.getProperty("jdk.net.preferIPv6Addresses"));
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl(node.getJdbcUrl());
        config.setUsername(user);
        config.setDriverClassName(driver);
        config.setPassword(pwd);
        config.setLeakDetectionThreshold(1000000);
        //10s
        config.setConnectionTimeout(10000);
        //30s
        config.setIdleTimeout(30000);
        //30s
        config.setMaxLifetime(30000);
        config.setMaximumPoolSize(8);
        config.setMinimumIdle(1);
         if(MonitorMethodEnum.ORALCE_JDBC_TPL.code().equals(me)){
             config.addDataSourceProperty("oracle.net.disableOob", "true");
             config.addDataSourceProperty("oracle.net.dns.usePlainAddress", "true");
             config.addDataSourceProperty("oracle.net.connect_timeout", "5000");
             config.addDataSourceProperty("oracle.jdbc.ReadTimeout", "30000");
             config.addDataSourceProperty("oracle.jdbc.useThreadLocalBufferCache", "true");
             config.addDataSourceProperty("oracle.jdbc.fanEnabled", "false"); // 禁用FAN事件
             config.addDataSourceProperty("oracle.jdbc.implicitStatementCacheSize", "0"); // 禁用FAN事件
             config.addDataSourceProperty("oracle.net.keepAlive", "false"); // 禁用FAN事件
             config.addDataSourceProperty("oracle.net.tcpNoDelay", "true"); // 禁用FAN事件
         }
        //设置为0，一开始不进行初始化
        config.setInitializationFailTimeout(0);
        HikariDataSource dataSource = null;
        Logger.info("node:"+node.getNodeNameShow()+",driver:"+driver);
        try {
            dataSource = new HikariDataSource(config);
            Logger.info("node:"+node.getNodeNameShow()+",start to getConnection,jdbc:"+node.getJdbcUrl());
            Connection conn=dataSource.getConnection();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            Logger.info("create ds failed!,url:"+node.getJdbcUrl());
            //插入节点报错数据
            MonitorTplIndicator indicator=monitorTplIndicatorList.get(0);
            Insert errInsert=new Insert("ops_monitor_node_value");
            errInsert.set("id", IDGenerator.getSnowflakeId());
            errInsert.setIf("indicator_code",indicator.getCode());
            errInsert.setIf("node_id",node.getId());
            if("db.connect".equals(indicator.getCode())){
                errInsert.setIf("is_connected",0);
                errInsert.setIf("result_status","sucess");
                errInsert.setIf("result_message","ds create failed");
            }else{
                errInsert.setIf("is_connected",1);
                errInsert.setIf("result_status","failed");
                errInsert.setIf("result_message","ds create failed");
            }
            errInsert.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
            errInsert.setIf("record_time",new Date());
            dao.execute(errInsert);
            errInsert.into("ops_monitor_node_value_last");
            dao.execute(errInsert);
            return ErrorDesc.failureMessage("ds创建失败");
        }
        JdbcTemplate jdbcTemplate = new JdbcTemplate();
        jdbcTemplate.setDataSource(dataSource);
        List<Insert> list=queryIndicators(jdbcTemplate,monitorTplIndicatorList,node);
        if(list.size()>0){
            monitorDataProcessBaseService.executeCollectDataInsert(list,"collect by jdbc tpl");
        }
        Logger.info("node:"+node.getNodeNameShow()+",finish");
        if (dataSource != null && !dataSource.isClosed()) {
            dataSource.close();
            Logger.info(node.getJdbcUrl()+",HikariDataSource已关闭，所有连接已释放。");
        }
        return ErrorDesc.success();
    }

    public List<Insert> queryIndicators(JdbcTemplate d,List<MonitorTplIndicator> indicators,MonitorNode node){
        List<Insert> insertList=new ArrayList<>();
        for(int i=0;i<indicators.size();i++){
            List<Insert> ins=queryIndicatorData(d,indicators.get(i),node);
            if(ins!=null&&ins.size()>0){
                insertList.addAll(ins);
            }
        }
        return insertList;
    }
    public List<Insert> queryIndicatorData(JdbcTemplate dbDao,MonitorTplIndicator indicator,MonitorNode node){
        List<Insert> list=new ArrayList<>();
        if(!indicator.getStatus().equals("enable")){
            return list;
        }
        List<Map<String, Object>>  rs=null;
        try{
            //{varList:['db','type']},用来定义具体的变量名称
            String indVar=indicator.getIndicatorVariable();
            //从node只获取具体的变量名称
            String var=node.getVar();
            String cmd=indicator.getCommand();
            if(StringUtil.isBlank(var)||StringUtil.isEmpty(indVar)){
                Logger.info("node:"+node.getNodeNameShow()+",var is none,indicator:"+indicator.getIndicatorVariable()+",var is blank!");
            }else{
                JSONObject varJson = JSONObject.parseObject(var);
                JSONObject indVarObj= JSONArray.parseObject(indVar);
                JSONArray indVarJson= new JSONArray();
                if(indVarObj.containsKey("varList")){
                    indVarJson=indVarObj.getJSONArray("varList");
                }
                if(indVarJson.size()>0&&varJson!=null){
                    for(int i=0;i<indVarJson.size();i++){
                        String key=indVarJson.getString(i).trim();
                        if(varJson.containsKey(key)){
                            String keyValue=varJson.getString(key);;
                            cmd=cmd.replace("${"+key+"}",keyValue);
                        }
                    }
                }
            }

            //只能执行select、show开头的命令
            boolean ckCmdOK=false;
            String chkCmdMsg="";
            if(StringUtil.isBlank(cmd)){
                ckCmdOK=false;
                chkCmdMsg="执行内容为空";
            }else{
                if(cmd.trim().toLowerCase().startsWith("select")
                        ||cmd.trim().toLowerCase().startsWith("call")
                        ||cmd.trim().toLowerCase().startsWith("show")
                ){
                    chkCmdMsg="检查通过";
                    ckCmdOK=true;
                }else{
                    chkCmdMsg="执行内容正确";
                    ckCmdOK=false;
                }
            }
            if(!ckCmdOK){
                Insert valueInsert2=new Insert("ops_monitor_node_value");
                valueInsert2.set("id", IDGenerator.getSnowflakeId());
                valueInsert2.setIf("result_status","failed");
                valueInsert2.setIf("result_message",chkCmdMsg);
                valueInsert2.setIf("indicator_code",indicator.getCode());
                valueInsert2.setIf("node_id",node.getId());
                valueInsert2.setIf("is_connected",1);
                valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                valueInsert2.setIf("record_time",new Date());
                list.add(valueInsert2);
                return list;
            }

            try {
                rs = dbDao.queryForList(cmd);
            }catch (EmptyResultDataAccessException e) {
                rs=null;
                Logger.error("node:"+node.getNodeNameShow()+",indicator:"+indicator.getCode()+",未找到对应记录");
            }
            if(rs==null||rs.size()==0){
                Logger.error("node:"+node.getNodeNameShow()+",indicator:"+indicator.getCode()+",query result size null");
                Insert valueInsert=new Insert("ops_monitor_node_value");
                valueInsert.set("id", IDGenerator.getSnowflakeId());
                valueInsert.setIf("result_status","failed");
                valueInsert.setIf("result_message","result is null");
                valueInsert.setIf("indicator_code",indicator.getCode());
                valueInsert.setIf("node_id",node.getId());
                valueInsert.setIf("is_connected",1);
                valueInsert.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                valueInsert.setIf("record_time",new Date());
                list.add(valueInsert);
            }else{
                String[] mapArr=indicator.getValueColumnMap().split(",");
                String content="";
                boolean dataCheck=true;
                String dataCheckMsg="";
                for(int i=0;i<rs.size();i++){
                    String e="";
                    Map<String,Object> itemRes=rs.get(i);
                    for(int j=0;j<mapArr.length;j++){
                        String v="";
                        String key=mapArr[j];
                        if(itemRes.containsKey(key)){
                            Object objectValue=itemRes.get(key);
                            if(objectValue instanceof String){
                                v=(String)objectValue;
                            }else if(objectValue instanceof BigDecimal){
                                v=(((BigDecimal)objectValue)).toString();
                            }else if(objectValue instanceof String){
                                v=(String)objectValue;
                            }else if(objectValue instanceof Long){
                                v=Long.toString((Long)objectValue);
                            }else if(objectValue instanceof Float){
                                v=Float.toString((Float)objectValue);
                            }else if(objectValue instanceof Integer){
                                v=Integer.toString((Integer)objectValue);
                            }else if(objectValue instanceof Date){
                                 Date d=(Date)objectValue;
                                 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                 v=sdf.format(d);
                            }else if(objectValue instanceof Short){
                                v=Short.toString((Short)objectValue);
                            }else if(objectValue instanceof Boolean){
                                v=String.valueOf((Boolean)objectValue);
                            }else if(objectValue instanceof BigInteger) {
                                v = String.valueOf((BigInteger) objectValue);
                            }else if(objectValue instanceof JSONObject) {
                                v =((JSONObject) objectValue).toJSONString();
                            }else if(objectValue instanceof JSONArray) {
                                v =((JSONArray) objectValue).toJSONString();
                            } else{
                                v=String.valueOf(objectValue);
                            }
                        }else{
                            dataCheckMsg="column key "+key+" not exist.";
                            Logger.info(dataCheckMsg);
                            dataCheck=false;
                            break;
                        }
                        if(j==0){
                            e=v;
                        }else{
                            e=e+","+v;
                        }
                    }
                    if(i==0){
                        content=e;
                    }else{
                        content=content+"\n"+e;
                    }
                }
                if(dataCheck){
                    Logger.info("content:\n"+content);
                    Result<Object> insertResult=monitorDataProcessBaseService.convertToInsertData(indicator,content,node);
                    if(insertResult.isSuccess()){
                        list=(List<Insert>)insertResult.getData();
                    }else{
                        Insert valueInsert2=new Insert("ops_monitor_node_value");
                        valueInsert2.set("id", IDGenerator.getSnowflakeId());
                        valueInsert2.setIf("result_status","failed");
                        valueInsert2.setIf("result_message",insertResult.getMessage());
                        valueInsert2.setIf("indicator_code",indicator.getCode());
                        valueInsert2.setIf("node_id",node.getId());
                        valueInsert2.setIf("is_connected",1);
                        valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                        valueInsert2.setIf("record_time",new Date());
                        list.add(valueInsert2);
                    }
                }else{
                    Insert valueInsert2=new Insert("ops_monitor_node_value");
                    valueInsert2.set("id", IDGenerator.getSnowflakeId());
                    valueInsert2.setIf("result_status","failed");
                    valueInsert2.setIf("result_message",dataCheckMsg);
                    valueInsert2.setIf("indicator_code",indicator.getCode());
                    valueInsert2.setIf("node_id",node.getId());
                    valueInsert2.setIf("is_connected",1);
                    valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                    valueInsert2.setIf("record_time",new Date());
                    list.add(valueInsert2);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            Insert valueInsert2=new Insert("ops_monitor_node_value");
            valueInsert2.set("id", IDGenerator.getSnowflakeId());
            valueInsert2.setIf("result_status","failed");
            valueInsert2.setIf("result_message",e.getMessage().length()>1024?e.getMessage().substring(0,1000):e.getMessage());
            valueInsert2.setIf("indicator_code",indicator.getCode());
            valueInsert2.setIf("node_id",node.getId());
            valueInsert2.setIf("is_connected",1);
            valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
            valueInsert2.setIf("record_time",new Date());
            list.add(valueInsert2);
        }
        return list;
    }

}
